package com.stanwind.sync.listener;

import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.google.protobuf.InvalidProtocolBufferException;
import com.stanwind.sync.SyncContext;
import com.stanwind.sync.anno.Listener;
import com.stanwind.sync.convert.DataConvertor;
import com.stanwind.sync.event.AbstractCanalEvent;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;

/**
 * RowDataSync Canal binlog sync listener
 *
 * @author : Stan
 * @version : 1.0
 * @date :  2019-08-07 11:19
 **/
public abstract class AbstractCanalListener<EVENT extends AbstractCanalEvent> implements ApplicationListener<EVENT> {

    private static final Logger logger = LoggerFactory.getLogger(AbstractCanalListener.class);

    private static final String DEFAULT_PK = "id";

    @Autowired
    protected DataConvertor dataConvertor;

    @Autowired
    protected ApplicationContext applicationContext;

    @Override
    public void onApplicationEvent(EVENT event) {
        Entry entry = event.getEntry();
        String database = entry.getHeader().getSchemaName();
        String table = entry.getHeader().getTableName();
        RowChange change;
        try {
            change = RowChange.parseFrom(entry.getStoreValue());
        } catch (InvalidProtocolBufferException e) {
            logger.error("canalEntry_parser_error,根据CanalEntry获取RowChange失败！", e);
            return;
        }
        change.getRowDatasList().forEach(rowData -> doSync(database, table, rowData));
    }

    /**
     * 使用新ID
     *
     * @param listener
     * @param rowData
     * @param beforePK 使用数据库变更前的数据
     * @param index    主键配置索引
     */
    protected void execListener(Listener listener, RowData rowData, boolean beforePK, int index) {
        this.initLazyContext(rowData, beforePK);
        Map<String, Object> map = beforePK ? SyncContext.getCurrentDataBefore()
                : SyncContext.getCurrentDataAfter();
        Object bean = applicationContext.getBean(listener.getClaz());
        String key = listener.getKeys().size() > index ? listener.getKeys().get(index) : DEFAULT_PK;
        Object param = map.get(key);
        logger.debug("exec sync method: {}, param {}={}", listener.getMethod().getName(), key, param);
        ReflectionUtils.invokeMethod(listener.getMethod(), bean, param);
    }

    /**
     * 把数据加入上下文 lazy转换before和after
     */
    protected void initLazyContext(RowData rowData, boolean beforeData) {
        SyncContext.clear();
        SyncContext.setCurrentListener(this);
        SyncContext.setCurrentIsDataBefore(beforeData);
        SyncContext.setCurrentRowData(rowData);

        if (beforeData) {
            SyncContext.setCurrentDataBefore(parseColumnsToMap(rowData.getBeforeColumnsList()));
        } else {
            SyncContext.setCurrentDataAfter(parseColumnsToMap(rowData.getAfterColumnsList()));
        }
    }

    /**
     * 获取默认变更行参数
     *
     * @param key
     * @param clazz
     * @param <T>
     */
    public <T> T getParam(String key, Class<T> clazz) {
        T t = SyncContext.getParam(key, clazz);
        //不存在时 根据rowData重新转换 以免lazy没置入数据
        if (t == null) {
            RowData rowData = SyncContext.getCurrentRowData();
            //默认已经装载了before或after之一的数据，现在加载另一部分
            if (!SyncContext.getCurrentIsDataBefore()) {
                SyncContext.setCurrentDataBefore(parseColumnsToMap(rowData.getBeforeColumnsList()));
            } else {
                SyncContext.setCurrentDataAfter(parseColumnsToMap(rowData.getAfterColumnsList()));
            }

            t = SyncContext.getParam(key, clazz);
        }

        return (T) t;
    }

    public <T> T getBeforeParam(String key, Class<T> clazz) {
        T t = SyncContext.getBeforeParam(key, clazz);
        //不存在时 根据rowData重新转换 以免lazy没置入数据
        if (t == null && !SyncContext.getCurrentIsDataBefore()) {
            RowData rowData = SyncContext.getCurrentRowData();
            //默认已经装载了before的数据，现在加载另一部分
            SyncContext.setCurrentDataBefore(parseColumnsToMap(rowData.getBeforeColumnsList()));

            t = SyncContext.getBeforeParam(key, clazz);
        }

        return t;
    }

    public <T> T getAfterParam(String key, Class<T> clazz) {
        T t = SyncContext.getAfterParam(key, clazz);
        //不存在时 根据rowData重新转换 以免lazy没置入数据
        if (t == null && SyncContext.getCurrentIsDataBefore()) {
            RowData rowData = SyncContext.getCurrentRowData();
            //默认已经装载了after的数据，现在加载另一部分
            SyncContext.setCurrentDataAfter(parseColumnsToMap(rowData.getAfterColumnsList()));

            t = SyncContext.getAfterParam(key, clazz);
        }

        return t;
    }


    /**
     * 数据行转换
     *
     * @param columns
     */
    protected Map<String, Object> parseColumnsToMap(List<Column> columns) {
        Map<String, Object> jsonMap = new HashMap<>();
        columns.forEach(column -> {
            if (column == null) {
                return;
            }
            jsonMap.put(column.getName(), column.getIsNull() ? null
                    : dataConvertor.getElasticsearchTypeObject(column.getMysqlType(), column.getValue()));
        });

        return jsonMap;
    }


    /**
     * 是否能够执行
     */
    protected boolean canExec(Listener listener, String database, String table) {
        //DB可为空
        return (StringUtils.isEmpty(listener.getDb()) || listener.getDb().equals(database))
                && listener.getTables().contains(table);
    }

    /**
     * 同步方法
     *
     * @param database
     * @param table
     * @param rowData
     */
    protected abstract void doSync(String database, String table, RowData rowData);
}
